import com.sun.org.apache.bcel.internal.generic.SWITCH;

import java.io.*;
import java.sql.*;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;
import javax.sql.DataSource;

/**
 * @作者 ： cwx951593
 * @时间： 2021/3/30
 **/
public class Import_tool_v3_peak {
    static public int batch_insert_records_per_time = 500;
    static public int batch_merge_recodrds_per_time = 250;
    static public String split_flag = ",";
    static public List<String> tb_inserts = null;
    static public List<String> tb_merges = null;
    static public List<String> dateFiles; //数据文件，必须与插入的文件保持一致
    public static volatile AtomicLong nInsertTables = new AtomicLong(0); //用于线程级别的计数器，保证一个线程一个插入操作
    public static volatile AtomicLong nMergetables = new AtomicLong(0);//用于线程级别的计数器，保证一个线程一个merge操作
    public static int numberOfTables = 0;
    public static volatile AtomicLong  memory_tracking_mode= new AtomicLong(0);
    static int threadsShutdown;

    public static void insert_to_table_tb_cddtl_cust_label_tmp(Connection conn, String tb_insert, String fileName) {
        File file = new File(fileName);
        BufferedReader reader = null;
        try {
            reader = new BufferedReader(new FileReader(file));//
            String tempString = null;
            int line = 1;
            int batchs = 0;
            String sql = String.format("merge into %s T using (select ? task_id, ? job_zone_no, ? temp_record_no,? data_date,? cust_no,? cust_label_cd,? valid_status_code,? is_skipped) as ab on (T.cust_no = ab.cust_no) WHEN MATCHED THEN  UPDATE SET " +
                    "T.task_id=ab.task_id,T.job_zone_no=ab.job_zone_no,T.temp_record_no=temp_record_no,T.data_date=ab.data_date," +
                    "T.cust_label_cd=ab.cust_label_cd,T.valid_status_code=ab.valid_status_code,T.is_skipped=ab.is_skipped WHEN NOT MATCHED THEN " +
                    "INSERT (task_id,job_zone_no,temp_record_no,data_date,cust_no,cust_label_cd,valid_status_code,is_skipped)" +
                    " values (ab.task_id,ab.job_zone_no,ab.temp_record_no,ab.data_date,ab.cust_no,ab.cust_label_cd,ab.valid_status_code,ab.is_skipped)", tb_insert);
            PreparedStatement ps_insert = conn.prepareStatement(sql);
            conn.setAutoCommit(false);
            Boolean flag = true;
            System.out.println("开始插入数据：" + tb_insert);
            while (flag) {
                for (int i = 0; i < batch_insert_records_per_time; i++) {
                    //此处设置值
                    if ((tempString = reader.readLine()) != null) {
                        String[] l = tempString.split(split_flag);
                        try {
                            ps_insert.setString(1, l[0]);
                            ps_insert.setInt(2, Integer.parseInt(l[1]));
                            ps_insert.setInt(3, Integer.parseInt(l[2]));
                            ps_insert.setString(4, l[3]);
                            ps_insert.setString(5, l[4]);
                            ps_insert.setString(6, l[5]);
                            ps_insert.setString(7, l[6]);
                            ps_insert.setInt(8, Integer.parseInt(l[7]));
                            line++;
                            ps_insert.addBatch();
                        } catch (SQLException e) {
                            System.out.println("insert error:" + Arrays.toString(l));
                            e.printStackTrace();
                        }
                    } else {
                        if (i == 0) {
                            //一条数据也没有加，直接退出
                            reader.close();
                            System.out.println("总计行数：" + (line - 1));
                            System.out.println("总计批处理数：" + (batchs));
                            return;
                        }
                        flag = false;
                        break;
                    }
                }
                batchs++;
                ps_insert.executeBatch();
                conn.commit();
            }
            reader.close();
            System.out.println("总计行数：" + (line - 1));
            System.out.println("总计批处理数：" + (batchs));

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (Exception el) {

                }
            }
        }
    }

    public static void merge_to_table_tb_cddtl_cust_label(Connection conn, Connection conn2, String tb_insert, String tb_merge) {
        String[] names = tb_insert.split("_");
        String zone_val = names[names.length - 1];
        try {
            int batchs = 0;
            //先查询
            conn2.setAutoCommit(false);//改为非自动提交，使得setFetchSize生效
            Statement st = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.FETCH_FORWARD);
            st.setFetchSize(batch_merge_recodrds_per_time);
            ResultSet rs = st.executeQuery(String.format("select cust_no,cust_label_cd,data_date from %s;", tb_insert));
            String sql = String.format("merge into %s T using (select ? cust_no, ? cust_label_cd, ? data_date) as ab on (T.cust_no = ab.cust_no) WHEN MATCHED THEN UPDATE SET " +
                    "T.new_core_cust_label=ab.cust_label_cd,T.data_date=ab.data_date," +
                    "T.last_mod_stamp=sysdate,T.optimist_lock_ver_no=66 WHEN NOT MATCHED THEN " +
                    "INSERT (cust_no,new_core_cust_label,data_date,create_stamp,last_mod_stamp,optimist_lock_ver_no,zone_val)" +
                    " values (ab.cust_no,ab.cust_label_cd,ab.data_date,sysdate,sysdate,66,'%s')", tb_merge, zone_val);
            PreparedStatement ps_merge = conn2.prepareStatement(sql);
            //开启监控
            if (memory_tracking_mode.getAndAdd(1)==0) {
                Statement st_set = conn2.createStatement();
                st_set.executeUpdate("set memory_tracking_mode=peak;");
                conn2.commit();
                st_set.close();
                System.out.println("对表："+tb_insert+"进行memory_tracking_mode=peak监控。");
            }
            int records = 0;
            Boolean flag = true;
            System.out.println("开始merge数据：" + tb_merge);
            while (flag) {
                for (int i = 0; i < batch_merge_recodrds_per_time; i++) {
                    if (rs.next()) {
                        ps_merge.setString(1, rs.getString(1));
                        ps_merge.setString(2, rs.getString(2));
                        ps_merge.setString(3, rs.getString(3));
                        //System.out.println(ps_merge);//liumin
                        records++;
                        ps_merge.addBatch();
                        //System.out.println(ps_merge);//liumin
                    } else {
                        if (i == 0) {
                            //一条数据也没有加，直接退出
                            System.out.println("总记录数：" + (records));
                            System.out.println("总计批处理数：" + (batchs));
                            return;
                        }
                        flag = false;
                        break;
                    }
                }
                batchs++;
                ps_merge.executeBatch();
                conn2.commit();
                //sleep(200000);
            }
            System.out.println("总记录数：" + (records));
            System.out.println("总计批处理数：" + (batchs));

        } catch (Exception e) {
            e.printStackTrace();
        } finally {

        }
    }
    public static void insert_to_table_tb_cddtl_cust_label_tmp_and_merge(Connection conn,String tb_insert,String tb_merge,String fileName) {
        File file=new File(fileName);
        BufferedReader reader = null;
        String[] names = tb_insert.split("_");
        String zone_val = names[names.length - 1];
        try {
            reader = new BufferedReader(new FileReader(file));//
            String tempString = null;
            String[] insertSql=new String[batch_insert_records_per_time];
            String[][] mergeValues=new String[batch_insert_records_per_time][3];
            int line = 1;
            int batchs = 0;
            String sql = String.format("merge into %s T using (select ? task_id, ? job_zone_no, ? temp_record_no,? data_date,? cust_no,? cust_label_cd,? valid_status_code,? is_skipped) as ab on (T.cust_no = ab.cust_no) WHEN MATCHED THEN  UPDATE SET " +
                    "T.task_id=ab.task_id,T.job_zone_no=ab.job_zone_no,T.temp_record_no=temp_record_no,T.data_date=ab.data_date," +
                    "T.cust_label_cd=ab.cust_label_cd,T.valid_status_code=ab.valid_status_code,T.is_skipped=ab.is_skipped WHEN NOT MATCHED THEN " +
                    "INSERT (task_id,job_zone_no,temp_record_no,data_date,cust_no,cust_label_cd,valid_status_code,is_skipped)" +
                    " values (ab.task_id,ab.job_zone_no,ab.temp_record_no,ab.data_date,ab.cust_no,ab.cust_label_cd,ab.valid_status_code,ab.is_skipped)", tb_insert);
            PreparedStatement ps_insert = conn.prepareStatement(sql);
            conn.setAutoCommit(false);
            Boolean flag = true;
            System.out.println("开始插入数据：" + tb_insert);
            while (flag) {
                for (int i = 0; i < batch_insert_records_per_time; i++) {
                    //此处设置值
                    if ((tempString = reader.readLine()) != null) {
                        String[] l = tempString.split(split_flag);
                        try {
                            ps_insert.setString(1, l[0]);
                            ps_insert.setInt(2, Integer.parseInt(l[1]));
                            ps_insert.setInt(3, Integer.parseInt(l[2]));
                            ps_insert.setString(4, l[3]);
                            ps_insert.setString(5, l[4]);
                            ps_insert.setString(6, l[5]);
                            ps_insert.setString(7, l[6]);
                            ps_insert.setInt(8, Integer.parseInt(l[7]));
                            line++;
                            ps_insert.addBatch();
                            mergeValues[i]= new String[]{l[4], l[5], l[3]};
                        } catch (SQLException e) {
                            System.out.println("insert error:" + Arrays.toString(l));
                            e.printStackTrace();
                        }
                    } else {
                        if (i == 0) {
                            //一条数据也没有加，直接退出
                            reader.close();
                            System.out.println("总计行数：" + (line - 1));
                            System.out.println("总计批处理数：" + (batchs));
                            return;
                        }
                        flag = false;
                        break;
                    }
                }
                batchs++;
                ps_insert.executeBatch();
                conn.commit();

                //插入完成直接merge
                //System.out.println("开始merge数据：" + tb_merge);
                String sql2 = String.format("merge into %s T using (select ? cust_no, ? cust_label_cd, ? data_date) as ab on (T.cust_no = ab.cust_no) WHEN MATCHED THEN UPDATE SET " +
                        "T.new_core_cust_label=ab.cust_label_cd,T.data_date=ab.data_date," +
                        "T.last_mod_stamp=sysdate,T.optimist_lock_ver_no=66 WHEN NOT MATCHED THEN " +
                        "INSERT (cust_no,new_core_cust_label,data_date,create_stamp,last_mod_stamp,optimist_lock_ver_no,zone_val)" +
                        " values (ab.cust_no,ab.cust_label_cd,ab.data_date,sysdate,sysdate,66,'%s')", tb_merge, zone_val);
                try{
                    PreparedStatement ps_merge = conn.prepareStatement(sql2);
                int count=1;
                boolean isExcuteAll=false;
                for(String[] k:mergeValues) {
                        isExcuteAll=false;
                        ps_merge.setString(1, k[0]);
                        ps_merge.setString(2, k[1]);
                        ps_merge.setString(3, k[2]);
                        ps_merge.addBatch();
                        System.out.println(ps_merge);
                        if(count%batch_merge_recodrds_per_time==0){
                            ps_merge.executeBatch();
                            conn.commit();
                            isExcuteAll=true;
                        }
                        count++;
                    }
                if (isExcuteAll){ //防止有部分没有执行。
                    ps_merge.executeBatch();
                    System.out.println(ps_merge);
                    conn.commit();
                }
                }catch (SQLException sqle){
                    sqle.printStackTrace();
                }
            }
        }catch (Exception e) {
            e.printStackTrace();
        }finally {
            if(reader != null) {
                try {
                    reader.close();
                }catch (Exception el) {

                }
            }
        }
    }
    public static void printUsage() {
        System.out.println("Usage:");
        System.out.println("-h --help: 帮助信息");
        System.out.println("jdbc连接参数：");
        System.out.println("-F --config-file: jdbc连接配置文件路径（可选，不指定此参数时，其它jdbc参数必须指定，带默认值的除外）");
        System.out.println("-u --user-name: jdbc连接用户名（可选）");
        System.out.println("-W --password: jdbc连接用户密码（可选）");
        System.out.println("-d --dbname: 导入数据数据库的名称");
        System.out.println("-s --schema-name: 导入数据模式名称(default public)");
        System.out.println("数据导入合并参数:");
        System.out.println("操作模式说明，必选其一：");
        System.out.println("--insert: 只进行导入操作:");
        System.out.println("    -f --data-file: 导入数据文件路径");
        System.out.println("    -S --split-flag: 表数据中的分隔符(default ,)");
        System.out.println("    -ti --table-name-insert: insert表名称(default tb_cddtl_cust_label_tmp),多个表用逗号分隔，或者用-，用短横时，自动解析末尾的数字，自动生成表名称");
        System.out.println("    -tm --table-name-merge: merge表名称(default tb_cddtl_cust_label)");
        System.out.println("    -bi --batch-insert-size: 一次批处理插入数据的条数（default 500)");
        System.out.println("--merge: 只进行merge操作:");
        System.out.println("    -ti --table-name-insert: insert表名称(default tb_cddtl_cust_label_tmp)");
        System.out.println("    -tm --table-name-merge: merge表名称(default tb_cddtl_cust_label)");
        System.out.println("    -bm --batch-merge-size: 一次批处理merge数据的条数（default 250)");
        System.out.println("--insert_merge: 先insert再merge操作（需指定--insert --merge的所有参数）");
        System.out.println("--threads: 最大线程并发数，默认值20");
        System.out.println("--max_jdbc_connections: 最大jdbc连接池连接数，默认值130");
        System.out.println("--memory_tracking_mode: 是否开启memory_tracking_mode，0：关闭，1开启。默认值1");
    }

    public static List<String> praseTables(String tables) {
        String[] tablesArr = tables.split(",");
        ArrayList<String> tablesList = new ArrayList<>();
        for (String t : tablesArr) {
            if (t.contains("-")) {
                String[] singleTable = t.split("-");
                if (singleTable.length != 2) {
                    System.out.println("表格式输入不正确，-表示自动生成表后缀数字的上限和下限。");
                    return new ArrayList<>();
                }
                String[] names = singleTable[0].split("_");
                int zone_val_start = Integer.parseInt(names[names.length - 1]);
                int suffixLen = names[names.length - 1].length(); //tb_cddtl_cust_label_tmp_0002的0002长度
                String tablePrefix = singleTable[0].substring(0, singleTable[0].length() - suffixLen); //tb_cddtl_cust_label_tmp_0002的tb_cddtl_cust_label_tmp_
                String[] names2 = singleTable[1].split("_");
                int zone_val_end = Integer.parseInt(names2[names2.length - 1]);
                for (int number = zone_val_start; number <= zone_val_end; number++) {
                    //后缀位数的补充，前位补充0
                    String tableNmber = String.format("%0" + suffixLen + "d", number);
                    tablesList.add(tablePrefix + tableNmber);
                }

            } else {
                tablesList.add(t);
            }
        }
        return tablesList;
    }

    public static void main(String[] args) {
        Properties pros = new Properties();
        Properties pros_conf = new Properties();
        String jdbc_config_file = null;
        String data_file = null;
        String operation_mode = null;
        String tb_insert = null;
        String tb_merge = null;
        int threads = 20;
        int max_jdbc_connections = 130;
        if (args.length <= 3) {
            printUsage();
            return;
        }
        for (int i = 0; i < args.length; i++) {
            switch (args[i]) {
                case "--insert":
                    operation_mode = "insert";
                    break;
                case "--merge":
                    operation_mode = "merge";
                    break;
                case "--insert_merge":
                    operation_mode = "insert_merge";
                    break;
                case "-bi":
                    batch_insert_records_per_time = Integer.parseInt(args[i + 1]);
                    break;
                case "--batch-insert-size":
                    batch_insert_records_per_time = Integer.parseInt(args[i + 1]);
                    break;
                case "-bm":
                    batch_merge_recodrds_per_time = Integer.parseInt(args[i + 1]);
                    break;
                case "--batch-merge-size":
                    batch_merge_recodrds_per_time = Integer.parseInt(args[i + 1]);
                    break;
                case "--config-file":
                    jdbc_config_file = args[i + 1];
                    pros_conf.putAll(Conn_Util.getConfigFromFile(jdbc_config_file));
                    break;
                case "-F":
                    jdbc_config_file = args[i + 1];
                    pros_conf.putAll(Conn_Util.getConfigFromFile(jdbc_config_file));
                    break;
                case "--data-file":
                    data_file = args[i + 1];
                    break;
                case "-f":
                    data_file = args[i + 1];
                    break;
                case "-d":
                    pros.put("dbname", args[i + 1]);
                    break;
                case "--dbname":
                    pros.put("dbname", args[i + 1]);
                    break;
                case "-ti":
                    tb_insert = args[i + 1];
                    break;
                case "--table-name-insert":
                    tb_insert = args[i + 1];
                    break;
                case "-tm":
                    tb_merge = args[i + 1];
                    break;
                case "--table-name-merge":
                    tb_merge = args[i + 1];
                    break;
                case "-s":
                    pros.put("currentSchema", args[i + 1]);
                    break;
                case "--schema-name":
                    pros.put("currentSchema", args[i + 1]);
                    break;
                case "-S":
                    split_flag = args[i + 1];
                    break;
                case "--split-flag":
                    split_flag = args[i + 1];
                    break;
                case "-u":
                    pros.put("user", args[i + 1]);
                    break;
                case "--user-name":
                    pros.put("user", args[i + 1]);
                    break;
                case "-W":
                    pros.put("password", args[i + 1]);
                    break;
                case "--password":
                    pros.put("password", args[i + 1]);
                    break;
                case "--threads":
                    threads = Integer.parseInt(args[i + 1]);
                    break;
                case "--max_jdbc_connections":
                    max_jdbc_connections = Integer.parseInt(args[i + 1]);
                    break;
                case "-h":
                    printUsage();
                    return;
                case "--help":
                    printUsage();
                    return;
                default:
                    break;
                case "--memory_tracking_mode":
                    int flag=Integer.parseInt(args[i + 1]);
                    if (flag==0){
                        memory_tracking_mode.getAndAdd(1); //增加1，相当于设置了一个，后续就不会再设置了。
                    }
            }
        }
        //以参数为准，覆盖配置文件中的
        pros_conf.putAll(pros);
        if (operation_mode == null) {
            System.out.println("必须指定参数--insert or --merge or --insert_merge.");
            return;
        }
        //使用连接池
        if (max_jdbc_connections > 0) {
            pros.put("useConnectionPools", "true");
            Conn_Util.setMaxPoolConnections(max_jdbc_connections);
        }
        //业务流程控制，启用线程池
        ExecutorService execService = Executors.newFixedThreadPool(threads);
        //每一个插入数据的业务都开启一个线程
        tb_inserts = praseTables(tb_insert);
        tb_merges = praseTables(tb_merge);
        dateFiles = praseTables(data_file);
        if(tb_inserts.isEmpty()||tb_merges.isEmpty()||dateFiles.isEmpty()){
            System.out.println("插入表和合入表和数据文件不对应。");
            return;
        }
        if (tb_merges.size() == 1 || (tb_merges.size() == tb_inserts.size())) {

        } else {
            System.out.println("merge表必须是一个或者与插入表一次对应。");
        }
        numberOfTables = tb_inserts.size();
        switch (operation_mode) {
            case "insert_merge":
                for (int i = 0; i < numberOfTables; i++) {
                    execService.execute(new Runnable() {
                        @Override
                        public void run() {
                            String threadname = null;
                            int thisTableIndex = 0;
                            try {
                                thisTableIndex = (int) nInsertTables.addAndGet(1) - 1;
                                //System.out.println(tb_inserts.toString());
                                threadname = tb_inserts.get(thisTableIndex);
                                System.out.println("启动插入合并线程：" + threadname + "_insert_merge_" + thisTableIndex + "，已启动插入合并线程数：" + (thisTableIndex + 1));
                                Thread.currentThread().setName(threadname + "_insert_merge_" + thisTableIndex);
                                insert_to_table_tb_cddtl_cust_label_tmp(Conn_Util.openDB(pros_conf, "importtools"),
                                        threadname, dateFiles.get(thisTableIndex));
                                if (tb_merges.size() == 1) {
                                    merge_to_table_tb_cddtl_cust_label(Conn_Util.openDB(pros_conf, "importtools"),
                                            Conn_Util.openDB(pros_conf, "importtools"),
                                            tb_inserts.get(thisTableIndex), tb_merges.get(0));
                                } else {
                                    merge_to_table_tb_cddtl_cust_label(Conn_Util.openDB(pros_conf, "importtools"),
                                            Conn_Util.openDB(pros_conf, "importtools"),
                                            tb_inserts.get(thisTableIndex), tb_merges.get(thisTableIndex));
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            } finally {
                                System.out.println("插入合并线程：" + threadname + "_insert_merge_" + thisTableIndex + "结束。");
                                threadsShutdown++;
                                //关闭线程池
                                if (threadsShutdown >= tb_inserts.size()) {
                                    execService.shutdown();
                                }
                            }
                        }
                    });
                }
                break;
            case "insert":
                for (int i = 0; i < numberOfTables; i++) {
                    execService.execute(new Runnable() {
                        @Override
                        public void run() {
                            String threadname = null;
                            int thisTableIndex = 0;
                            try {
                                thisTableIndex = (int) nInsertTables.addAndGet(1) - 1;
                                System.out.println(tb_inserts.toString());
                                threadname = tb_inserts.get(thisTableIndex);
                                System.out.println("启动插入线程：" + threadname + "_insert_" + thisTableIndex + "，已启动插入线程数：" + (thisTableIndex + 1));
                                Thread.currentThread().setName(threadname + "_insert_" + thisTableIndex);

                                insert_to_table_tb_cddtl_cust_label_tmp(Conn_Util.openDB(pros_conf, "importtools"),
                                        threadname, dateFiles.get(thisTableIndex));
                            } catch (Exception e) {
                                e.printStackTrace();
                            } finally {
                                System.out.println("插入线程：" + threadname + "_insert_" + thisTableIndex + "结束。");
                                threadsShutdown++;
                                //关闭线程池
                                if (threadsShutdown >= tb_inserts.size()) {
                                    execService.shutdown();
                                }
                            }

                        }
                    });
                }

                break;
            //merge
            case "merge":
                for (int i = 0; i < numberOfTables; i++) {
                    execService.execute(new Runnable() {
                        @Override
                        public void run() {
                            int thisTableIndex = 0;
                            try {
                                thisTableIndex = (int) nMergetables.addAndGet(1) - 1;
                                Thread.currentThread().setName(tb_inserts.get(thisTableIndex) + "_merge_" + thisTableIndex);
                                System.out.println("启动merge线程：" + tb_inserts.get(thisTableIndex) + "_merge_" + thisTableIndex + "，已启动merge线程数：" + (thisTableIndex + 1));
                                if (tb_merges.size() == 1) {
                                    merge_to_table_tb_cddtl_cust_label(Conn_Util.openDB(pros_conf, "importtools"),
                                            Conn_Util.openDB(pros_conf, "importtools"),
                                            tb_inserts.get(thisTableIndex), tb_merges.get(0));
                                } else {
                                    merge_to_table_tb_cddtl_cust_label(Conn_Util.openDB(pros_conf, "importtools"),
                                            Conn_Util.openDB(pros_conf, "importtools"),
                                            tb_inserts.get(thisTableIndex), tb_merges.get(thisTableIndex));
                                }
                            } catch (Exception e) {
                                e.printStackTrace();
                            } finally {
                                System.out.println("merge线程：" + tb_inserts.get(thisTableIndex) + "_merge_" + thisTableIndex + "结束。");
                                threadsShutdown++;
                                //关闭线程池
                                if (threadsShutdown >= tb_inserts.size()) {
                                    execService.shutdown();
                                }
                            }

                        }
                    });
                }
                break;
            default:
                System.out.println("没有此模式：" + operation_mode);
        }
        //关闭线程池
        if (threadsShutdown >= tb_inserts.size()) {
            execService.shutdown();
        }
    }
}